package com.smartrodin.config;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.commons.lang3.concurrent.BasicThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.util.CollectionUtils;
import javax.annotation.PostConstruct;
import java.util.Calendar;
import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * Author: arnold_nie
 * Date: 2020/4/26 9:21
 * FileName: RedisAbstractDelayQueue
 * Description: redis 延迟队列
 */
public abstract class AbstractRedisDelayQueue {

    protected Logger logger = LoggerFactory.getLogger(AbstractRedisDelayQueue.class);

    @Autowired
    protected RedisTemplate redisTemplate;

    /**
     * 插入任务id
     * @param taskId 任务id(队列内唯一)
     * @param time 延时时间(单位 :秒)
     * @return 是否插入成功
     */
    public boolean addTaskId(String taskId,  Integer time) {
        Calendar instance = Calendar.getInstance();
        instance.add(Calendar.SECOND, time);
        long delaySeconds = instance.getTimeInMillis() / 1000;
        try {
            String key = setDelayQueueName();
            boolean zadd = redisTemplate.opsForZSet().add(key, taskId, delaySeconds);
            return zadd;
        } catch (Exception e) {
            e.printStackTrace();
            return false;
        }
    }

    private void startDelayQueue() {
        logger.info("AbstractRedisDelayQueue---init----QueueName==> {}", setDelayQueueName());
        while (true) {
            try {
                long now = System.currentTimeMillis() / 1000;
                // 获取当前时间前的任务列表
                Set<Object> taskIds = redisTemplate.opsForZSet().rangeByScoreWithScores(setDelayQueueName(), 0, now);
                // 如果不为空则遍历判断其是否满足取消要求
                if (!CollectionUtils.isEmpty(taskIds)) {
                    for (Object obj : taskIds) {
                        JSONObject jsonObject = JSONObject.parseObject(JSON.toJSONString(obj));
                        String taskId = jsonObject.get("value").toString();
                        Long num = redisTemplate.opsForZSet().remove(setDelayQueueName(), taskId);
                        // 如果移除成功, 则取消订单
                        if (null != num && num > 0) {
                            invoke(taskId);
                        }
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
                logger.warn("处理延时任务发生异常,异常原因为 {}", e.getMessage(), e);
            } finally {
                // 间隔一秒钟执行一次
                try {
                    TimeUnit.SECONDS.sleep(1L);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }

            }
        }
    }


    /**
     * 最终执行的任务方法
     * @param taskId 任务id
     */
    public abstract void invoke(String taskId);

    /**
     * 要实现延时队列的名字
     *
     */
    public abstract String setDelayQueueName();

    /**
     * 项目启动时执行初始化
     */
    @PostConstruct
    public void init(){
        ExecutorService singleThreadExecutor = new ThreadPoolExecutor(1, 1, 300L, TimeUnit.SECONDS, new LinkedBlockingQueue<Runnable>(16),
                new BasicThreadFactory.Builder().namingPattern("start-delay-queue-thread-pool-%d").daemon(true).build(), new ThreadPoolExecutor.AbortPolicy());
        singleThreadExecutor.execute(()-> {
            logger.info("startDelayQueue--- {}", Thread.currentThread().getName());
            startDelayQueue();
        });
        singleThreadExecutor.shutdown();
    }

}
